package com.ptone.messageCenter.api;

import com.ptone.messageCenter.common.response.ErrCode;
import com.ptone.messageCenter.common.response.Result;
import com.ptone.messageCenter.common.response.Tuple2;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.admin.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.Arrays;
import java.util.Properties;
import java.util.Set;

@RestController
public class TopicAPI {

    @Value("${topic.bootstrap.servers}")
    private String kafkaUrl;
    @Value("${topic.sessionTimeOut}")
    private int sessionTimeOut;
    @Value("${topic.connectionTimeOut}")
    private int connectionTimeOut;

    // 创建主题
    @PostMapping("/createTopic")
    private Result createTopic(@RequestParam("topicName") String topicName, @RequestParam("partition") Integer partition, @RequestParam("replicationFactor") short replicationFactor) {
        Properties properties = new Properties();
        properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, kafkaUrl);
        AdminClient adminClient = AdminClient.create(properties);
        NewTopic newTopic = new NewTopic(topicName, partition, replicationFactor);
        CreateTopicsResult topics = adminClient.createTopics(Arrays.asList(newTopic));
        try {
            topics.all().get();
            System.out.println("主题：[" + topicName + "]创建成功!");
            return Result.end(Tuple2.left(ErrCode.OK));
        } catch (Exception e) {
            e.printStackTrace();
            return Result.end(Tuple2.right(ErrCode.FAIL));
        } finally {
            adminClient.close();
        }
    }

    // 查询所有主题
    @PostMapping("/selectAllTopics")
    private Result selectTopics() {
        AdminClient adminClient = null;
        try {
            Properties properties = new Properties();
            properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, kafkaUrl);
            adminClient = AdminClient.create(properties);
            ListTopicsOptions options = new ListTopicsOptions();
            options.listInternal(false); // 是否包括自带主题， as __consumer_offsets
            ListTopicsResult listTopicsResult = adminClient.listTopics(options);
            Set<String> strings = listTopicsResult.names().get();
            return Result.end(Tuple2.left(strings));
        } catch (Exception e) {
            e.printStackTrace();
            return Result.end(Tuple2.right(ErrCode.FAIL));
        } finally {
            adminClient.close();
        }
    }

    // 查询所有主题
    @PostMapping("/selectTopics")
    private Result selectTopics(@RequestParam String topicNames) {
        AdminClient adminClient = null;
        DeleteTopicsResult deleteTopicsResult;
        try {
            Properties properties = new Properties();
            properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, kafkaUrl);
            adminClient = AdminClient.create(properties);
            deleteTopicsResult = adminClient.deleteTopics(Arrays.asList(topicNames));
            return Result.end(Tuple2.left(deleteTopicsResult));
        } catch (Exception e) {
            e.printStackTrace();
            return Result.end(Tuple2.right(ErrCode.FAIL));
        } finally {
            adminClient.close();
        }
    }
}
